package com.xxxx.spark.operator

import com.xxxx.spark.WorldCount
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子是返回一个新的RDD
 */
object TransformationsFun {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local").setAppName("transformation_operator")
    val sc = new SparkContext(conf)

    //数据(RDD)的产生，RDD之间的相互转换
    //val file = WorldCount.getClass.getClassLoader.getResource("src/main/resources/data/hello.txt").getFile

   // val lineRDD = sc.textFile(file)
    //println(sc.defaultMinPartitions)
   // println(sc.defaultParallelism)

    /**
     * 下列算子研究并行度问题
     */
    /**
     * join
     * 产生新rdd的分区数继承大的rdd
     * 并行度=partition的个数=task的个数
     */
    /*val rdd1: RDD[(Int, String)] = sc.parallelize(List(
      Tuple2(0, "aa"),
      Tuple2(1, "spd"),
      (2, "sdf"),
      (4, "rtr"),
    ),4)
    val rdd2: RDD[(Int, Int)] = sc.parallelize(List(
      Tuple2(0, 23),
      Tuple2(1, 76),
      (2, 80),
      (3, 99),
    ),2)

    val rdd3: RDD[(Int, (String, Int))] = rdd1.join(rdd2)
    rdd3.foreach(println)
    val rdd4: RDD[(Int, (String, Option[Int]))] = rdd1.leftOuterJoin(rdd2)
    val rdd5: RDD[(Int, (Option[String], String))] = rdd1.rightOuterJoin(rdd1)
    val rdd6: RDD[(Int, (Option[String], Option[Int]))] = rdd1.fullOuterJoin(rdd2)

    println(rdd3.getNumPartitions)
    println(rdd3.partitions.size)*/

    /**
     * union
     * 子rdd并行度为父rdd并行度之和
     */
    /*val rdd1: RDD[Int] = sc.parallelize(List(1,2,3,4),4)
    val rdd2: RDD[Int] = sc.parallelize(List(4,5,6),2)

    val rdd3: RDD[Int] = rdd2.union(rdd1)
    println(rdd3.getNumPartitions)//6*/

    /**
     * 交集 intersection
     * 子rdd的并行度为父rdd中并行度大的那个
     */
    /*val rdd4: RDD[Int] = rdd1.intersection(rdd2)
    rdd4.foreach(println)//4
    println(rdd4.getNumPartitions)//4*/

    /**
     * 差集 subtract
     * 子rdd的并行度为父rdd中并行度大的那个
     */
    /*val rdd5: RDD[Int] = rdd1.subtract(rdd2)
    rdd5.foreach(print)//1,2,3
    println("----")
    println(rdd5.getNumPartitions)//4*/

    /**
     * mapPartitions
     * 和数据库相关的操作（外存->走io了）
     * 类似于批量处理--->效率比较高
     * 注意批量处理的内存问题
     * 决定因素是partition的设定，有几批
     */
   /* val rdd = sc.parallelize(List(1, 2, 3, 4), 3)
    rdd.mapPartitions(iter =>{
      println("创建数据库连接")
      val list = List()
      while(iter.hasNext){
        val i = iter.next()
        println("查询到数据库中数据"+ i)
        i +: list
      }
      println("关闭数据库连接")
      list.iterator
    }).count()

    rdd.map(x => {
      println("创建数据库连接")
      println("查询到数据库中数据"+ x)
      println("关闭数据库连接")
      x
    }).count()*/

    /**
     *foreachPartition
     * 同上
     */
    /*val rdd = sc.parallelize(List(1, 2, 3, 4,5,6,7,8,9), 3)
    rdd.foreachPartition(iter =>{
      println("创建数据库连接")
      val list = List()
      while(iter.hasNext){
        val i = iter.next()
        println("查询到数据库中数据"+ i)
        i +: list
      }
      println("关闭数据库连接")
      list.iterator
    })

    rdd.foreach(x => {
      println("创建数据库连接")
      println("查询到数据库中数据"+ x)
      println("关闭数据库连接")
      x
    })*/

    /**
     * distinct
     * ?有哪些操作可以改变并行度
     * reduceByKey
     * textFile
     * distinct
     * 如何实现？
     * map + reduceByKey(num) + map
     */
    val rdd = sc.parallelize(List(1, 2, 3, 3, 3, 3, 8, 8, 4, 9), 3)
    rdd.distinct(4).foreach(println)

    sc.stop()
  }
}
